package com.lanmaoly.demo;

import java.io.*;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 给定两个文件，分别有100亿个整数，只提供1G内存，如何找出两文件交集？
 *
 * @author fangzhe.hou 2019-11-4 14:49:51
 */
public class CalculateSet {

    private static final int HASHLEN = 100;
    private static final int FORI = HASHLEN << 1;

    public static void main(String[] args) throws IOException {
        // 实际上不能用多线程 内存不够  40G/100 1G 400M max=2
//        ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS,
//                new LinkedBlockingQueue<>(), new NamedThreadFactory("split-file"), new ThreadPoolExecutor.CallerRunsPolicy());
//        String path = "D:/set/";
        String path = "D:/split/";
        String largeFilePath1 = path + "largefile1.txt";
        String largeFilePath2 = path + "largefile2.txt";
        String splitPath1 = path + "split1";
        String splitPath2 = path + "split2";
        String intersectionFilePath = path + "/intersection/intersection.txt";

        // 先将文件进行分割，将文件中的整数hash取模存放
        fileSplit(largeFilePath1, splitPath1);
        fileSplit(largeFilePath2, splitPath2);
        // 读取各自分割之后的文件
        File file1 = new File(splitPath1);
//        String[] fileNames1 = file1.list();
        File[] files1 = file1.listFiles();

        File file2 = new File(splitPath2);
//        String[] fileNames2 = file2.list();
        File[] files2 = file2.listFiles();
        calculate(intersectionFilePath, files1, files2);
        // calculateSet(intersectionFilePath, files1, files2);
        System.out.println("----------结束----------");
    }

    private static void calculate(String intersectionFilePath, File[] files1, File[] files2) throws IOException {
        if (files1 != null && files2 != null) {
            File intersectionFile = new File(intersectionFilePath);
            FileWriter fw = new FileWriter(intersectionFile, true);
            PrintWriter pw = new PrintWriter(fw);

            Map<String, File> fileMap1 = Arrays.stream(files1).
                    collect(Collectors.toMap(File::getName, file -> file, (f1, f2) -> f1));
            Map<String, File> fileMap2 = Arrays.stream(files2).
                    collect(Collectors.toMap(File::getName, file -> file, (f1, f2) -> f1));

            Iterator<Map.Entry<String, File>> iterator1 = fileMap1.entrySet().iterator();
            while (iterator1.hasNext()) {
                Map.Entry<String, File> fileEntry = iterator1.next();
                File splitFile1 = fileEntry.getValue();
                String fileName = fileEntry.getKey();
                File splitFile2 = fileMap2.get(fileName);
                if (splitFile2 != null) {
                    BufferedReader reader1 = new BufferedReader(new FileReader(splitFile1));
                    BufferedReader reader2 = new BufferedReader(new FileReader(splitFile2));
                    Set<String> set1 = new HashSet<>();
                    Set<String> set2 = new HashSet<>();
                    String line1 = reader1.readLine();
                    while (null != line1) {
                        set1.add(line1);
                        line1 = reader1.readLine();
                    }
                    String line2 = reader2.readLine();
                    while (null != line2) {
                        set2.add(line2);
                        line2 = reader2.readLine();
                    }
                    reader1.close();
                    reader2.close();
                    // 两个文件重复数据写入一个单独的文件
                    set1.retainAll(set2);
                    Iterator<String> iterator = set1.iterator();
                    while (iterator.hasNext()) {
                        pw.println(iterator.next());
                        pw.flush();
                        iterator.remove();
                    }
                }
                iterator1.remove();
            }
            fw.flush();
            pw.close();
            fw.close();
        }
    }

    private static void calculateSet(String intersectionFilePath, File[] files1, File[] files2) throws IOException {
        if (files1 != null && files2 != null) {
            File intersectionFile = new File(intersectionFilePath);
            FileWriter fw = new FileWriter(intersectionFile, true);
            PrintWriter pw = new PrintWriter(fw);
            for (int i = 0; i < FORI; i++) {
                if (files1.length > i && files2.length > i) {
                    File splitFile1 = files1[i];
                    File splitFile2 = files2[i];
                    BufferedReader reader1 = new BufferedReader(new FileReader(splitFile1));
                    BufferedReader reader2 = new BufferedReader(new FileReader(splitFile2));
                    Set<String> set1 = new HashSet<>();
                    Set<String> set2 = new HashSet<>();
                    String line1 = reader1.readLine();
                    while (null != line1) {
                        set1.add(line1);
                        line1 = reader1.readLine();
                    }
                    String line2 = reader2.readLine();
                    while (null != line2) {
                        set2.add(line2);
                        line2 = reader2.readLine();
                    }
                    reader1.close();
                    reader2.close();
                    // 两个文件重复数据写入一个单独的文件
                    set1.retainAll(set2);
                    Iterator<String> iterator = set1.iterator();
                    while (iterator.hasNext()) {
                        String next = iterator.next();
                        pw.println(next);
                        pw.flush();
                        iterator.remove();
                    }
                } else {
                    System.out.println("----------break-----------");
                    break;
                }
            }
            fw.flush();
            pw.close();
            fw.close();
        }
    }

    private static void fileSplit(String largeFilePath, String splitPath) throws IOException {
        BufferedReader reader = new BufferedReader(new FileReader(largeFilePath));
        // 读取一行
        String line = reader.readLine();
        while (null != line) {
            // 计算hash
            final int hash = hash(line.toCharArray());
//            System.out.println(hash);
            // 保存文件
            File file = new File(splitPath + File.separator + hash + ".txt");
            if (file.exists()) {
                boolean newFile = file.createNewFile();
            }
            FileWriter fw = new FileWriter(file, true);
            PrintWriter pw = new PrintWriter(fw);
            pw.println(line);
            pw.flush();
            fw.flush();
            fw.close();
            pw.close();
            // 读取一行
            line = reader.readLine();
        }
        reader.close();
    }

    private static int hash(char[] word) {
        int index = 0;
        int i = 0;
        while (i < word.length) {
            index += index * 31 + word[i];
            i++;
        }
        return index % HASHLEN;
    }
}
